package com.tmt.im.manager.config;

import com.tmt.im.common.bean.DataSourceConfig;
import com.tmt.im.common.pojo.*;
import com.tmt.im.common.store.*;
import com.tmt.im.manager.helper.MySqlHelper;
import lombok.extern.slf4j.Slf4j;
import org.apache.ignite.*;
import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.QueryEntity;
import org.apache.ignite.cache.store.CacheStoreAdapter;
import org.apache.ignite.configuration.CacheConfiguration;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.jdbc.DataSourceProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.StringUtils;

import javax.cache.configuration.Factory;
import javax.cache.configuration.FactoryBuilder;
import javax.cache.expiry.CreatedExpiryPolicy;
import javax.cache.expiry.ExpiryPolicy;
import java.sql.*;
import java.text.MessageFormat;
import java.time.Duration;
import java.time.OffsetDateTime;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.concurrent.TimeUnit;

/**
 * Ignite配置
 *
 * @Description TODO
 * @Author wuyi
 * @Date 2024/7/16 15:18
 * @Version 1.0
 **/

@Slf4j
@Configuration
public class IgniteConfig {

    private final DataSourceProperties dataSourceProperties;

    @Value("${ignite.Chat-data.expire-days:3}")
    private int expireDays;

    private final String igniteConfigFile;

    @Value("${oauth2.token-expire-time:PT300S}")
    private Duration tokenExpireTime;

    public IgniteConfig(@Value("${ignite.config-xml:classpath:config/ignite-config.xml}")
                        String igniteConfigFile,
                        DataSourceProperties dataSourceProperties) {
        this.igniteConfigFile = igniteConfigFile;
        this.dataSourceProperties = dataSourceProperties;
    }

    @Bean
    public Ignite igniteInstance() {
        Ignite ignite = Ignition.start(igniteConfigFile);
        //使用全局信号量确保以下初始化过程仅被执行一次。
        try (IgniteSemaphore semaphore = ignite.semaphore("systemSyn", 1, true, true)) {
            if (semaphore.tryAcquire(3, TimeUnit.SECONDS)) {
                try {
                    IgniteCache<String, Object> configParamCache = ignite.cache(Constant.APPLICATION_CONFIG_NAME);
                    if (configParamCache == null) {
                        //创建全局参数缓存。
                        configParamCache = ignite.getOrCreateCache(Constant.APPLICATION_CONFIG_NAME);
                    }

                    Object obj = configParamCache.get(Constant.DATA_SOURCE_CONFIG_NAME);
                    if (obj == null) {
                        DataSourceConfig config = new DataSourceConfig(dataSourceProperties.getUrl(),
                                dataSourceProperties.getUsername(), dataSourceProperties.getPassword());
                        //将数据源配置放入全局参数缓存中。
                        configParamCache.put(Constant.DATA_SOURCE_CONFIG_NAME, config);
                    }

                    IgniteCache<String, Object> tmpCache = ignite.cache(Constant.LOGOUT_USER_CACHE_NAME);
                    if (tmpCache == null) {
                        //创建登出用户缓存。
                        CacheConfiguration<Long, String> cacheCfg = new CacheConfiguration<>();
                        cacheCfg.setName(Constant.LOGOUT_USER_CACHE_NAME);
                        Factory<ExpiryPolicy> expiryPolicy = CreatedExpiryPolicy.factoryOf(
                                new javax.cache.expiry.Duration(TimeUnit.SECONDS, tokenExpireTime.getSeconds()));
                        cacheCfg.setExpiryPolicyFactory(expiryPolicy);
                        ignite.getOrCreateCache(cacheCfg);
                    }

                    IgniteCache<String, Object> tmpCache1 = ignite.cache(Constant.INVALID_TOKEN_CACHE_NAME);
                    if (tmpCache1 == null) {
                        //创建废弃令牌缓存。
                        CacheConfiguration<Long, String> cacheCfg = new CacheConfiguration<>();
                        cacheCfg.setName(Constant.INVALID_TOKEN_CACHE_NAME);
                        Factory<ExpiryPolicy> expiryPolicy = CreatedExpiryPolicy.factoryOf(
                                new javax.cache.expiry.Duration(TimeUnit.SECONDS, tokenExpireTime.getSeconds()));
                        cacheCfg.setExpiryPolicyFactory(expiryPolicy);
                        ignite.getOrCreateCache(cacheCfg);
                    }

                    String initLoadSQL = null;
                    IgniteCache<Long, IMUserPoJo> cache = ignite.cache(IMUserPoJo.CACHE_NAME);
                    if (cache == null) {
                        //初始化用户对象缓存,并将现有会话记录加载到缓存中。
                        initLoadSQL = MessageFormat.format("select * from {0}", IMUserPoJo.CACHE_NAME);
                        initialCache(ignite, IMUserPoJo.CACHE_NAME, CacheAtomicityMode.TRANSACTIONAL, Long.class,
                                IMUserPoJo.class, IMUserStore.class, initLoadSQL, null);
                    }

                    IgniteCache<Long, ChatSessionPoJo> cache1 = ignite.cache(ChatSessionPoJo.CACHE_NAME);
                    if (cache1 == null) {
                        initLoadSQL = MessageFormat.format("select * from {0}", ChatSessionPoJo.CACHE_NAME);
                        //初始化聊天会话对象缓存,并将现有会话记录加载到缓存中。
                        initialCache(ignite, ChatSessionPoJo.CACHE_NAME, CacheAtomicityMode.TRANSACTIONAL, Long.class,
                                ChatSessionPoJo.class, ChatSessionStore.class, initLoadSQL, null);
                    }

                    IgniteCache<Long, ChatSessionMemberPoJo> cache2 = ignite.cache(ChatSessionMemberPoJo.CACHE_NAME);
                    if (cache2 == null) {
                        //初始化聊天成员对象缓存,并将现有会话成员记录加载到缓存中。
                        initLoadSQL = MessageFormat.format("select * from {0}", ChatSessionMemberPoJo.CACHE_NAME);
                        initialCache(ignite, ChatSessionMemberPoJo.CACHE_NAME, CacheAtomicityMode.TRANSACTIONAL, SessionMemberKey.class,
                                ChatSessionMemberPoJo.class, ChatSessionMemberStore.class, initLoadSQL, null);
                    }

                    Timestamp validTime = Timestamp.from(OffsetDateTime.now().minusDays(expireDays).toInstant());

                    IgniteCache<Long, ChatDataPoJo> cache3 = ignite.cache(ChatDataPoJo.CACHE_NAME);
                    if (cache3 == null) {
                        //初始化聊天数据对象缓存,并将现有未过期的聊天记录加载到缓存中。
                        initLoadSQL = MessageFormat.format("select * from {0} where create_time > ?", ChatDataPoJo.CACHE_NAME);
                        initialCache(ignite, ChatDataPoJo.CACHE_NAME, CacheAtomicityMode.TRANSACTIONAL, ChatDataKey.class,
                                ChatDataPoJo.class, ChatDataStore.class, initLoadSQL, new Object[]{validTime});
                    }

                    IgniteCache<Long, IMFriendPoJo> cache4 = ignite.cache(IMFriendPoJo.CACHE_NAME);
                    if (cache4 == null) {
                        initLoadSQL = MessageFormat.format("select * from {0}", IMFriendPoJo.CACHE_NAME);
                        initialCache(ignite, IMFriendPoJo.CACHE_NAME, CacheAtomicityMode.TRANSACTIONAL, IMFriendKey.class,
                                IMFriendPoJo.class, IMFriendStore.class, initLoadSQL, null);
                    }

                    IgniteCache<Long, IMFriendApplyPoJo> cache5 = ignite.cache(IMFriendApplyPoJo.CACHE_NAME);
                    if (cache5 == null) {
                        //先删除过期数据
                        String sql = MessageFormat.format("delete from {0} where review_time is null and create_time < ?", IMFriendApplyPoJo.CACHE_NAME);
                        MySqlHelper.executeBy(sql, new Object[]{validTime}, dataSourceProperties);
                        //再将有效数据加载到内存中.
                        initLoadSQL = MessageFormat.format("select * from {0}", IMFriendApplyPoJo.CACHE_NAME);
                        initialCache(ignite, IMFriendApplyPoJo.CACHE_NAME, CacheAtomicityMode.TRANSACTIONAL, Long.class,
                                IMFriendApplyPoJo.class, IMFriendApplyStore.class, initLoadSQL, null);
                    }

                    IgniteCache<Long, IMFriendGroupPoJo> cache6 = ignite.cache(IMFriendGroupPoJo.CACHE_NAME);
                    if (cache6 == null) {
                        initLoadSQL = MessageFormat.format("select * from {0}", IMFriendGroupPoJo.CACHE_NAME);
                        initialCache(ignite, IMFriendGroupPoJo.CACHE_NAME, CacheAtomicityMode.TRANSACTIONAL, Long.class,
                                IMFriendGroupPoJo.class, IMFriendGroupStore.class, initLoadSQL, null);
                    }

                    IgniteCache<Long, IMFriendGroupMappingPoJo> cache7 = ignite.cache(IMFriendGroupMappingPoJo.CACHE_NAME);
                    if (cache7 == null) {
                        initLoadSQL = MessageFormat.format("select * from {0}", IMFriendGroupMappingPoJo.CACHE_NAME);
                        initialCache(ignite, IMFriendGroupMappingPoJo.CACHE_NAME, CacheAtomicityMode.TRANSACTIONAL, Long.class,
                                IMFriendGroupMappingPoJo.class, IMFriendGroupMappingStore.class, initLoadSQL, null);
                    }

                    IgniteCache<BlackBillKey, BlackBillPoJo> cache8 = ignite.cache(BlackBillPoJo.CACHE_NAME);
                    if (cache8 == null) {
                        initLoadSQL = MessageFormat.format("select * from {0}", BlackBillPoJo.CACHE_NAME);
                        initialCache(ignite, BlackBillPoJo.CACHE_NAME, CacheAtomicityMode.TRANSACTIONAL, BlackBillKey.class,
                                BlackBillPoJo.class, BlackBillStore.class, initLoadSQL, null);
                    }

                    IgniteCache<Long, PersonalSettingPoJo> cache9 = ignite.cache(PersonalSettingPoJo.CACHE_NAME);
                    if (cache9 == null) {
                        initLoadSQL = MessageFormat.format("select * from {0}", PersonalSettingPoJo.CACHE_NAME);
                        initialCache(ignite, PersonalSettingPoJo.CACHE_NAME, CacheAtomicityMode.TRANSACTIONAL, Long.class,
                                PersonalSettingPoJo.class, PersonalSettingStore.class, initLoadSQL, null);
                    }

                    IgniteCache<Long, WalletPoJo> cache10 = ignite.cache(WalletPoJo.CACHE_NAME);
                    if (cache10 == null) {
                        initLoadSQL = MessageFormat.format("select * from {0}", WalletPoJo.CACHE_NAME);
                        initialCache(ignite, WalletPoJo.CACHE_NAME, CacheAtomicityMode.TRANSACTIONAL, Long.class,
                                WalletPoJo.class, WalletStore.class, initLoadSQL, null);
                    }

                    IgniteCache<Long, YunXinEventPoJo> cache11 = ignite.cache(YunXinEventPoJo.CACHE_NAME);
                    if (cache11 == null) {
                        initialCache(ignite, YunXinEventPoJo.CACHE_NAME, CacheAtomicityMode.ATOMIC, Long.class,
                                YunXinEventPoJo.class, YunXinEventStore.class, null, null);
                    }

                    IgniteCache<Long, YunXinEventPoJo> cache12 = ignite.cache(SystemNotifyPoJo.CACHE_NAME);
                    if (cache12 == null) {
                        initLoadSQL = MessageFormat.format("select * from {0}", SystemNotifyPoJo.CACHE_NAME);
                        initialCache(ignite, SystemNotifyPoJo.CACHE_NAME, CacheAtomicityMode.ATOMIC, SystemNotifyKey.class,
                                SystemNotifyPoJo.class, SystemNotifyStore.class, initLoadSQL, null);
                    }

                } catch (Exception ex) {
                    log.error("----Ignite数据初始化失败----" + ex.getMessage());
                }
            } else {
                log.error("获取全局信号量失败");
            }
        }
        return ignite;
    }

    private <K, T extends IGetKey<K>> IgniteCache<K, T> initialCache(Ignite ignite, String cacheName, CacheAtomicityMode mode,
                                                                     Class<K> keyClazz, Class<T> valueClazz,
                                                                     Class<? extends CacheStoreAdapter<K, T>> storeClass,
                                                                     String initLoadSQL, Object[] sqlParams) {
        CacheConfiguration<K, T> cacheCfg = new CacheConfiguration<>();
        cacheCfg.setName(cacheName);
        cacheCfg.setCacheMode(CacheMode.PARTITIONED);
        cacheCfg.setAtomicityMode(mode);
        cacheCfg.setReadThrough(true);
        cacheCfg.setWriteThrough(true);
        cacheCfg.setCacheStoreFactory(FactoryBuilder.factoryOf(storeClass));
        cacheCfg.setBackups(1);

        //获取字段定义列表.
        LinkedHashMap<String, String> fields = StoreSqlHelper.getFieldDefineList(valueClazz);
        QueryEntity queryEntity = new QueryEntity()
                .setTableName(cacheName)
                .setKeyFieldName("id").setKeyType(keyClazz.getName()).setValueType(valueClazz.getName())
                .setFields(fields);

        cacheCfg.setQueryEntities(Collections.singletonList(queryEntity));
        IgniteCache<K, T> cache = ignite.getOrCreateCache(cacheCfg);

        //加载所有数据。
        if (StringUtils.hasText(initLoadSQL)) {
            log.info("-------向{}缓存加载数据-----开始------", cacheName);
            loadAll(ignite, cacheName, valueClazz, initLoadSQL, sqlParams);
            log.info("-------向{}缓存加载数据-----结束------", cacheName);
        }
        return cache;
    }

    private <K, T extends IGetKey<K>> void loadAll(Ignite ignite, String cacheName, Class<T> clazz, String initLoadSQL, Object[] sqlParams) {
        //根据缓存名获取流处理器，并往流处理器中添加数据
        try (IgniteDataStreamer<K, T> streamer = ignite.dataStreamer(cacheName)) {
            try (Connection conn = DriverManager.getConnection(dataSourceProperties.getUrl(), dataSourceProperties.getUsername(),
                    dataSourceProperties.getPassword())) {
                try (PreparedStatement st = conn.prepareStatement(initLoadSQL)) {
                    //初始化SQL中的参数
                    if (sqlParams != null) {
                        MySqlHelper.initPreparedStatement(st, sqlParams);
                    }
                    ResultSet result = st.executeQuery();
                    while (result.next()) {
                        T obj = clazz.getDeclaredConstructor().newInstance();
                        StoreSqlHelper.initialObject(result, obj);
                        streamer.addData(obj.getKey(), obj);
                    }
                }
            } catch (Exception ex) {
                log.error(ex.getMessage(), ex);
            }
            //将流里面的剩余数据压进ignite
            streamer.flush();
        }
    }

}